package com.adcc.ags.utility.mq.transfer;

import com.adcc.ags.utility.log.Log;
import com.adcc.ags.utility.mq.configuration.MQConfiguration;
import com.adcc.ags.utility.mq.entity.Message;
import com.adcc.ags.utility.mq.entity.QMState;
import com.google.common.base.Optional;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.ibm.mq.*;
import com.ibm.mq.MQMessage;
import com.ibm.mq.constants.CMQC;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.List;

/**
 * MQTransfer
 */
public class MQTransferImpl extends MQTransfer implements AsyncMsgListener{

    // QueueManager(用于长连接QM)
    private MQQueueManager mqm = null;

    // 线程池对象
    protected ExecutorService executorService;

    /**
     * 构造器函数
     */
    public MQTransferImpl() {
        super();
    }

    /**
     * 构造函数
     * @param configuration
     */
    public MQTransferImpl(MQConfiguration configuration) {
        super(configuration);
    }

    /**
     * 构造函数
     * @param configuration
     * @param connectionPool
     */
    public MQTransferImpl(MQConfiguration configuration,MQConnectionPool connectionPool){
        super(configuration,connectionPool);
    }

    /**
     * 是否存在异步接收线程
     * @param name
     * @param receiverList
     * @return
     */
    private boolean isExistAsyncReceiver(String name,List<AsyncReceiver> receiverList) throws Exception{
        if(Strings.isNullOrEmpty(name)){
            throw new NullPointerException("AsyncReceiver name is null or empty");
        }
        if(receiverList.size() > 0){
            for(AsyncReceiver receiver : receiverList){
                if(receiver.equals(name)){
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 发布异步状态
     */
    private void notifyAsyncState(QMState state){
        this.asyncState = state;
        if(qmStateListener != null){
            qmStateListener.onState(configuration.getHost(),configuration.getPort(),configuration.getQueueManager(),asyncState);
        }
    }

    /**
     * 打开异步接收处理
     * @param mqm
     */
    private void openAsyncReceivers(MQQueueManager mqm){
        try{
            if(asyncQReceivers.size() > 0){
                for(AsyncReceiver receiver : asyncQReceivers){
                    receiver.setMqm(mqm);
                    receiver.open();
                }
            }
            if(asyncTReceivers.size() > 0){
                for(AsyncReceiver receiver : asyncTReceivers){
                    receiver.setMqm(mqm);
                    receiver.open();
                }
            }
        }catch (Exception ex){
            Log.error("openAsyncReceivers() error",ex);
        }
    }

    /**
     * 关闭异步接收处理
     */
    private void closeAsyncReceivers(){
        try{
            if(asyncQReceivers.size() > 0){
                for(AsyncReceiver receiver : asyncQReceivers){
                    receiver.close();
                }
            }
            if(asyncTReceivers.size() > 0){
                for(AsyncReceiver receiver : asyncTReceivers){
                    receiver.setMqm(mqm);
                    receiver.close();
                }
            }
        }catch (Exception ex){
            Log.error("closeAsyncReceivers() error",ex);
        }
    }

    @Override
    public void setQMStateListener(QMStateListener listener) {
        qmStateListener = listener;
    }

    @Override
    public void setQueueMsgListener(QueueMsgListener listener) {
        queueMsgListener = listener;
    }

    @Override
    public void setQueueMsgListener(QueueMsgListener listener, String...queue) throws Exception{
        if(queue != null && queue.length > 0){
            for (String q : queue){
                if(!isExistAsyncReceiver(q,asyncQReceivers)){
                    AsyncReceiver receiver = new AsyncQueueReceiver(configuration, q);
                    receiver.setListener(this);
                    asyncQReceivers.add(receiver);
                }
            }
        }
        queueMsgListener = listener;
    }

    @Override
    public void setTopicMsgListener(TopicMsgListener listener) {
        topicMsgListener = listener;
    }

    @Override
    public void setTopicMsgListener(TopicMsgListener listener, String...topic) throws Exception{
        if(topic != null && topic.length > 0){
            for (String t : topic){
                if(!isExistAsyncReceiver(t,asyncTReceivers)){
                    AsyncReceiver receiver = new AsyncTopicReceiver(configuration, t);
                    receiver.setListener(this);
                    asyncTReceivers.add(receiver);
                }
            }
        }
        topicMsgListener = listener;
    }

    @Override
    public boolean isConnectQM(){
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            if(mqm.isOpen() && mqm.isConnected() ){
                return true;
            }else{
                return false;
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"isConnectQM() error",ex);
            return false;
        }finally {
            try {
                returnQueueManager(mqm);
            } catch (Exception ex) {
                Log.error("close mq error",ex);
            }
        }
    }

    @Override
    public Optional<Message> receiveQueue(String queue) throws Exception{
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            int options = CMQC.MQOO_INPUT_AS_Q_DEF | CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            gmo.options = CMQC.MQGMO_NO_WAIT | CMQC.MQGMO_CONVERT | CMQC.MQGMO_FAIL_IF_QUIESCING;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen() && mqQueue.getCurrentDepth() > 0){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqQueue.get(mqMessage,gmo);
                int msgLength = mqMessage.getMessageLength();
                byte[] buffer = new byte[msgLength];
                mqMessage.readFully(buffer,0,buffer.length);
                Message message = new Message(buffer);
                return Optional.of(message);
            }else {
                return Optional.absent();
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"receive() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public Optional<Message> receiveQueue(String queue,String headKey) throws Exception {
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            int options = CMQC.MQOO_INPUT_AS_Q_DEF | CMQC.MQOO_INQUIRE | CMQC.MQOO_OUTPUT;
            gmo.options = CMQC.MQGMO_NO_WAIT | CMQC.MQGMO_CONVERT | CMQC.MQGMO_FAIL_IF_QUIESCING;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen() && mqQueue.getCurrentDepth() > 0){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqQueue.get(mqMessage,gmo);
                int msgLength = mqMessage.getMessageLength();
                Map<String,String> properties = Maps.newHashMapWithExpectedSize(1);
                properties.put(headKey,mqMessage.getStringProperty(headKey));
                byte[] buffer = new byte[msgLength];
                mqMessage.readFully(buffer,0,buffer.length);
                Message message = new Message(properties,buffer);
                return Optional.of(message);
            }else{
                return Optional.absent();
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"receive() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public Optional<Message> receiveQueue(String queue, Set<String> headKeys) throws Exception {
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            int options = CMQC.MQOO_INPUT_AS_Q_DEF | CMQC.MQOO_INQUIRE | CMQC.MQOO_OUTPUT;
            gmo.options = CMQC.MQGMO_NO_WAIT | CMQC.MQGMO_CONVERT | CMQC.MQGMO_FAIL_IF_QUIESCING;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen() && mqQueue.getCurrentDepth() > 0){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqQueue.get(mqMessage,gmo);
                int msgLength = mqMessage.getMessageLength();
                Map<String,String> properties = null;
                if (headKeys!= null && headKeys.size()>0) {
                    properties = Maps.newHashMapWithExpectedSize(headKeys.size());
                    for (String headKey : headKeys) {
                        properties.put(headKey, mqMessage.getStringProperty(headKey));
                    }
                }
                byte[] buffer = new byte[msgLength];
                mqMessage.readFully(buffer,0,buffer.length);
                Message message = new Message(properties,buffer);
                return Optional.of(message);
            }else{
                return Optional.absent();
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"receive() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public void sendQueue(String queue, byte[] msg) throws Exception{
        MQQueueManager mqm = null;
        try{
            mqm =  getQueueManager();
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            int options = CMQC.MQOO_OUTPUT | CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            pmo.options = CMQC.MQPMO_FAIL_IF_QUIESCING | CMQC.MQPMO_ASYNC_RESPONSE;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen()){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqMessage.write(msg,0,msg.length);
                mqQueue.put(mqMessage,pmo);
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"send() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public void sendQueue(String queue, byte[] msg,String headKey,String headValue) throws Exception{
        MQQueueManager mqm = null;
        try{
            mqm =  getQueueManager();
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            int options = CMQC.MQOO_OUTPUT | CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            pmo.options = CMQC.MQPMO_FAIL_IF_QUIESCING | CMQC.MQPMO_ASYNC_RESPONSE;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen()){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqMessage.write(msg,0,msg.length);
                mqMessage.setStringProperty(headKey, headValue);
                mqQueue.put(mqMessage,pmo);
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"send() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public void sendQueue(String queue, byte[] message, Map<String, String> properties) throws Exception {
        MQQueueManager mqm = null;
        try{
            mqm =  getQueueManager();
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            int options = CMQC.MQOO_OUTPUT | CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            pmo.options = CMQC.MQPMO_FAIL_IF_QUIESCING | CMQC.MQPMO_ASYNC_RESPONSE;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            if(mqQueue.isOpen()){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqMessage.write(message, 0, message.length);
                if (properties != null && properties.size() > 0) {
                    for (Map.Entry<String, String> entry : properties.entrySet()) {
                        mqMessage.setStringProperty(entry.getKey(), entry.getValue());
                    }
                }
                mqQueue.put(mqMessage,pmo);
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"send() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public int getDepth(String queue) throws Exception {
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            int options = CMQC.MQOO_INQUIRE | CMQC.MQOO_FAIL_IF_QUIESCING;
            MQQueue mqQueue = mqm.accessQueue(queue,options);
            int depth = mqQueue.getCurrentDepth();
            return depth;
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"send() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public void sendTopic(String topic, byte[] message,int expiry) throws Exception {
        MQQueueManager mqm = null;
        try{
            mqm = getQueueManager();
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            int options = CMQC.MQOO_OUTPUT | CMQC.MQOO_FAIL_IF_QUIESCING;
            pmo.options = CMQC.MQPMO_FAIL_IF_QUIESCING;
            MQTopic mqTopic = mqm.accessTopic(topic, topic, CMQC.MQTOPIC_OPEN_AS_PUBLICATION,options);
            if(mqTopic.isOpen()){
                MQMessage mqMessage = new MQMessage();
                mqMessage.characterSet = 1208;
                mqMessage.expiry = expiry;
                mqMessage.write(message,0,message.length);
                mqTopic.put(mqMessage,pmo);
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"sendToTopic() error",ex);
            throw ex;
        }finally {
            returnQueueManager(mqm);
        }
    }

    @Override
    public Optional<byte[]> receiveTopic(String topic) throws Exception {
        MQQueueManager mqm = null;
        byte[] buffer = null;
        try{
            mqm = getQueueManager();
            MQGetMessageOptions gmo = new MQGetMessageOptions();
            int options = CMQC.MQSO_CREATE;
            MQTopic mqTopic = mqm.accessTopic(topic,topic,CMQC.MQTOPIC_OPEN_AS_SUBSCRIPTION,options);
            if(mqTopic.isOpen()){
                MQMessage msg = new MQMessage();
                msg.characterSet = 1208;
                mqTopic.get(msg,gmo);
                int msgLength = msg.getMessageLength();
                buffer = new byte[msgLength];
                msg.readFully(buffer, 0, msgLength);
                return Optional.of(buffer);
            }else {
                return  Optional.absent();
            }
        }catch (Exception ex){
            Log.error(MQTransfer.class.getName(),"receiveTopic() error",ex);
            throw ex;
        }finally {
            if(mqm != null){
                returnQueueManager(mqm);
            }
        }
    }

    @Override
    public void startAsync() {
        try{
            try{
                mqm = getQueueManager();
            }catch (MQException ex){
                Log.error("get mqm instance error",ex);
            }
            executorService = Executors.newSingleThreadExecutor();
            executorService.execute(new MonitorTask(mqm));
        }catch (Exception ex){
            Log.error("startAsyncReceive() error",ex);
        }
    }

    @Override
    public void stopAsync() {
        try{
            closeAsyncReceivers();
            try{
                returnQueueManager(mqm);
            }catch (MQException ex){
                Log.error("return mqm instance error",ex);
            }
            if(executorService != null && !executorService.isShutdown()){
                executorService.shutdownNow();
            }
            Thread.sleep(1000);
            notifyAsyncState(QMState.Disconnect);
        }catch (Exception ex){
            Log.error("stopAsyncReceive() error",ex);
        }
    }

    @Override
    protected MQQueueManager getQueueManager() throws MQException{
        if(connectionPool == null){
            return new MQQueueManager(configuration.getQueueManager(),configuration.getParams());
        }else{
            return connectionPool.getQueueManager(configuration.getQueueManager(),configuration.getParams());
        }
    }

    @Override
    protected void returnQueueManager(MQQueueManager mqm) throws MQException{
        if(mqm != null){
            if(connectionPool == null){
                if(mqm.isConnected()){
                    mqm.disconnect();
                }
                if(mqm.isOpen()){
                    mqm.close();
                }
            }else{
                connectionPool.returnQueueManager(mqm);
            }
        }
    }

    @Override
    public void onMessage(String host,int port,String qm,String name,Message message,int type) {
        try{
            if(message != null){
                if(type == 0){
                    if(queueMsgListener != null){
                        queueMsgListener.onMessage(host,port,qm,name,message);
                    }
                }else{
                    if(topicMsgListener != null){
                        topicMsgListener.onMessage(host,port,qm,name,message);
                    }
                }
            }
        }catch (Exception ex){
            Log.error("onMessage() error",ex);
        }
    }

    /**
     * 监控线程
     */
    class MonitorTask implements Runnable{

        public MQQueueManager mqm;

        /**
         * 构造函数
         * @param mqm
         */
        public MonitorTask(MQQueueManager mqm){
            this.mqm = mqm;
        }

        @Override
        public void run() {
            try{
                while (true){
                    try{
                        MQAsyncStatus status = mqm.getAsyncStatus();
                        if(status.completionCode == 0){
                            if(status.reasonCode == CMQC.MQCC_OK || status.reasonCode == CMQC.MQRC_NO_MSG_AVAILABLE ||  status.reasonCode == CMQC.MQRC_OPTIONS_CHANGED){
                                if(asyncState != QMState.Connected){
                                    openAsyncReceivers(mqm);
                                    notifyAsyncState(QMState.Connected);
                                }
                            }else{
                                Log.error("connect qm " + configuration.getQueueManager() + " error reasonCode:" + status.reasonCode);
                                if(asyncState != QMState.Connecting){
                                    notifyAsyncState(QMState.Connecting);
                                }

                                // 关闭异步接收线程
                                closeAsyncReceivers();

                                // 重新获取QueueManager
                                try{
                                    returnQueueManager(mqm);
                                    mqm = getQueueManager();
                                }catch (MQException e){
                                    Log.error("get mqm instance error",e);
                                }
                            }
                        }else{
                            Log.error("connect qm " + configuration.getQueueManager() + " error reasonCode:" + status.reasonCode);
                            if(asyncState != QMState.Connecting){
                                notifyAsyncState(QMState.Connecting);
                            }

                            // 关闭异步接收线程
                            closeAsyncReceivers();

                            // 重新获取QueueManager
                            try{
                                returnQueueManager(mqm);
                                mqm = getQueueManager();
                            }catch (MQException e){
                                Log.error("get mqm instance error",e);
                            }
                        }
                    }catch (Exception ex){
                        Log.error("connect qm " + configuration.getQueueManager() + " error",ex);
                        if(asyncState != QMState.Connecting){
                            notifyAsyncState(QMState.Connecting);
                        }

                        // 关闭异步接收线程
                        closeAsyncReceivers();

                        // 重新获取QueueManager
                        try{
                            returnQueueManager(mqm);
                            mqm = getQueueManager();
                        }catch (MQException e){
                            Log.error("get mqm instance error",e);
                        }
                    }
                    Thread.sleep(interval);
                }
            }catch (Exception ex){
                Log.error("run () error");
            }
        }
    }
}
